package com.shujia.windowApply;
/*
 * 117988031603010,1614711895,36.38
 * 117988031603010,1614711904,35.38
 * 117988031603010,1614711914,45.38
 * 117988031603010,1614711924,45.29
 * 117988031603010,1614712022,75.29
 * 117988031603010,1614712120,46.29
 * 117988031603010,1614712218,82.29
 * 117988031603010,1614712316,82.29
 * 117988031603010,1614712414,48.5
 * 117988031603010,1614712619,59.5
 * 117988031603010,1614712824,52.5
 * 117988031603010,1614713030,71.5
 * 117988031603010,1614713235,65.5
 * 117988031603010,1614713441,78.26
 * 117988031603010,1614713547,76.26
 * 117988031603010,1614713653,108.26
 * 117988031603010,1614813653,108.26
 */

import com.shujia.process.Student;
import lombok.Data;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/* 判断卡口是否拥堵
 * 每隔1分钟统计最近15分钟卡口的平均车速
 */
public class CarWindowApply {
    public static void main(String[] args)throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> src = env.socketTextStream("hadoop102", 8888);

        DataStream<Car> cars = src.map(line -> new Car(line.split(",")));
        DataStream<Car> tswkDS = cars.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Car>forMonotonousTimestamps()
                        .withTimestampAssigner((car, l) -> car.getTs())
        );
        KeyedStream<Car, String> carsDS = tswkDS.keyBy(Car::getSid);
        WindowedStream<Car, String, TimeWindow> windowDS = carsDS.window(SlidingEventTimeWindows.of(Time.minutes(3), Time.minutes(1)));
        windowDS.apply(new WindowFunction<Car, Tuple2<String, Double>, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<Car> iterable, Collector<Tuple2<String, Double>> collector) throws Exception {
                Double sumSpeed = 0.0;
                Long count = 0L;
                for (Car car : iterable) {
                    sumSpeed += car.getSpeed();
                    count++;
                }
                Double avgSpeed = sumSpeed / count;

                collector.collect(Tuple2.of(s, avgSpeed));
            }
        }).print();
        env.execute();
    }
}
